#include "keyword_schema.h"
#include "keyword.h"
#include "schema_operate.h"
#include <assert.h>
#include "log.h"
#include "conf.h"
//#define MYSQL_KEYWORD_SCHEMA_SERVER 	"localhost"
//#define MYSQL_KEYWORD_SCHEMA_USER	 	"webroot"
//#define MYSQL_KEYWORD_SCHEMA_PASSWORD 	"123456"
//#define MYSQL_KEYWORD_SCHEMA_DB 		"keyword_schema"

keyword_schema_t g_keyword_schema;

void quit_pthread_keyword_schema(MYSQL* mysql, 
	pthread_mutex_t& locker, pthread_cond_t& cond )
{
	pthread_mutex_destroy(&locker);
	pthread_cond_destroy(&cond);
	mysql_close(mysql);
}

void* pthread_keyword_schema(void* data)
{
	pthread_detach(pthread_self());
	
	int i = (int)data;
	list<table_op_t*>& table_ops = g_keyword_schema.table_ops[i];
	list<table_op_t*>::iterator it;
	pthread_mutex_t& locker = g_keyword_schema.lockers[i];
	pthread_cond_t&  cond = g_keyword_schema.conds[i];
	
	MYSQL* mysql = mysql_init(NULL);
	if(!mysql_real_connect(mysql,
		g_conf.keyword.host.c_str(),
		g_conf.keyword.user.c_str(),
		g_conf.keyword.password.c_str(),
		g_conf.keyword.name.c_str(), 0, NULL, 0))
	{
		printf("error, mysql connect keyword_schema failed\n");
		return (void*)-1;
	}
	while(1)
	{
		pthread_mutex_lock(&locker);
		while(table_ops.size()==0)
		{
			pthread_cond_wait(&cond, &locker);
		}
		table_op_t* table_op = table_ops.front();
		table_ops.pop_front();
		pthread_mutex_unlock(&locker);
		
		if( table_op == 0)
			continue;
		
		eTABLE_OP op = table_op->op;
		schema_t* schema = (schema_t*)table_op->schema;
		keyword_t* keyword = (keyword_t*)schema->data;
		assert(keyword != NULL);
		// insert keyword to db ..
		int ret = 0;
		if( op == INSERT_TABLE)
		{
			ret = insert_to_keyword(mysql, keyword);	
		}
		else
		{
			assert(false);
		}
		
		
		go_next_schema(schema, ret);
		
		del_table_op(table_op);
		if(g_keyword_schema.quit)
		{
			pthread_mutex_lock(&locker);
			if(table_ops.size()==0)
			{
				pthread_mutex_unlock(&locker);
				break;
			}
			pthread_mutex_unlock(&locker);
		}
	}	
	quit_pthread_keyword_schema(mysql,locker, cond);
	return (void*)0;
}

int init_keyword_schema(keyword_schema_t& keyword_schema)
{
	
	keyword_schema.quit = false;
	for(int i = 0; i < KEYWORD_SCHEMA_THREAD_COUNT; i++)
	{
		pthread_mutex_init(&keyword_schema.lockers[i], NULL);
	}
	for(int i = 0; i < KEYWORD_SCHEMA_THREAD_COUNT; i++)
	{
		pthread_cond_init(&keyword_schema.conds[i], NULL);
	}
	pthread_t th_keyword_schema[KEYWORD_SCHEMA_THREAD_COUNT];
	for(int i = 0 ; i < KEYWORD_SCHEMA_THREAD_COUNT; i++)
	{
		pthread_create(&th_keyword_schema[i], NULL, &pthread_keyword_schema,
					   (void*)i);
	}
	return 0;
}

int op_keyword_schema(schema_t* schema,  eTABLE_OP op)
{
	if( schema == 0 )
		assert(false);
	//approximate the min	
	int index = rand()%KEYWORD_SCHEMA_THREAD_COUNT;
	if(index >= KEYWORD_SCHEMA_THREAD_COUNT)
	{
		index = KEYWORD_SCHEMA_THREAD_COUNT -1;
	}

	table_op_t* table_op = creat_table_op(schema, op);
	pthread_mutex_lock(&g_keyword_schema.lockers[index]);
	g_keyword_schema.table_ops[index].push_back(table_op);
	pthread_cond_signal(&g_keyword_schema.conds[index]);
	pthread_mutex_unlock(&g_keyword_schema.lockers[index]);	
	return 0;
}

int add_to_keyword_schema(void* data)
{
	assert(data != NULL);
	return op_keyword_schema((schema_t*)data, INSERT_TABLE);
}

int remove_from_keyword_schema(void* data)
{
	assert(data != NULL);
	return op_keyword_schema((schema_t*)data, DELETE_TABLE);
}

void quit_keyword_schema(keyword_schema_t* keyword_schema)
{
	keyword_schema->quit = true;
}
